package DianShang_2024.ds_03.feature

import org.apache.spark.ml.Pipeline
import org.apache.spark.ml.feature.{OneHotEncoder, StringIndexer, VectorAssembler}
import org.apache.spark.ml.linalg.{SparseVector, Vectors}
import org.apache.spark.sql.{SparkSession, functions}
import org.apache.spark.sql.functions.{array, col, collect_list, first, lit, max, substring, udf, when}
import spire.compat.fractional

import java.util.Properties

object feature_count {
  def main(args: Array[String]): Unit = {
    /*
          据Hive的dwd库中相关表或MySQL数据库shtd_store中订单相关表（order_detail、order_info、sku_info），对用户购买过的商品进行去重，
          将其转换为以下格式：第一列为用户id mapping，第二列为用户购买过的商品id mapping，按照user_id与sku_id进行升序排序，输出前5行，将结
          果截图粘贴至客户端桌面【Release\任务C提交结果.docx】中对应的任务序号下
     */

    val spark=SparkSession.builder()
      .master("local[*]")
      .appName("特征工程01")
      .config("hive.exec.dynamic.partition.mode","nonstrict")
//      .config("spark.serializer","org.apache.spark.serializer")
//      .config("spark.sql.extensions","org.apache.spark.sql.hudi.HoodieSparkSessionExtension")
      .enableHiveSupport()
      .getOrCreate()

    import spark.implicits._


    val connect=new Properties()
    connect.setProperty("user","root")
    connect.setProperty("password","123456")
    connect.setProperty("driver","com.mysql.jdbc.Driver")

    def read_mysql(mysql_name:String):Unit={
      val mysql_jdbc="jdbc:mysql://192.168.40.110:3306/shtd_store?useSSL=false"
      spark.read.jdbc(mysql_jdbc,mysql_name,connect)
        .createOrReplaceTempView(mysql_name)
    }

    read_mysql("sku_info")
    read_mysql("order_info")
    read_mysql("order_detail")

    /*
          1.剔除订单信息表与订单详细信息表中用户id与商品id不存在于现有的维表中的记录(就是将三张表连接起来，如果有用户一次都没有买过，那么就连接不起来
         也就达到了剔除的效果)
         2.对用户购买过的商品进行去重
         3.where t1.user_id is not null and t3.sku_id is not null这一条要加，可以起到筛选的效果帮助剔除
     */
    val r1=spark.sql(
      """
        |select distinct
        |t1.user_id,
        |t3.id
        |from order_info as t1
        |join order_detail as t2
        |on t2.order_id=t1.id
        |join sku_info as t3
        |on t3.id=t2.sku_id
        |where t1.user_id is not null and t3.id is not null
        |""".stripMargin)

//    r1.show

    //  定义StringIndexer用于映射(mapping),每一列都需要定义
    //  这里从机器学习的角度来说是注册模型，也就是测试模型
    val user_indexer=new StringIndexer()
      .setInputCol("user_id")                               //  选择改变的字段
      .setOutputCol("user_id_mapping")                      //  输出的字段
      .fit(r1)                                            //  根据r1来选择字段


    val sku_indexer=new StringIndexer()
      .setInputCol("id")
      .setOutputCol("sku_id_mapping")
      .fit(r1)


    //  执行映射(mapping)
    //  然后这里从机器学习的角度来说就是根据测试模型来训练模型
    val mapping_01=user_indexer.transform(r1)
    val result=sku_indexer.transform(mapping_01)


    println("-------user_id_mapping与sku_id_mapping数据前5条如下：-------")
    //  设置排序和选择字段
     result
      .withColumn("user_id_mapping",col("user_id_mapping").cast("int"))
      .withColumn("sku_id_mapping",col("sku_id_mapping").cast("int"))
      .select("user_id_mapping","sku_id_mapping")                         //  选择字段
      .orderBy("user_id_mapping","sku_id_mapping")                  //  排序规则
       .show(5,false)




    //      ______________第二题________________________
    /*
          根据第1小题的结果，对其进行聚合，其中对sku_id进行one-hot转换，将其转换为以下格式矩阵：第一列为用户id，其余列名为商品id，按照用户id进
          行升序排序，展示矩阵第一行前5列数据，将结果截图粘贴至客户端桌面【Release\任务C提交结果.docx】中对应的任务序号下

          one_hot编码:一行数据里面只有1和0,而且使用one-hot编码之前也要先映射，在one-hot
     */


    //  创建VectorAssembler来组合特征列
    /*
          VectorAssembler 是 SparkML 中的一个工具，用于将多个列组合成一个向量列，通常用于机器学习模型中的特征准备。在你的代码片段中
          ，VectorAssembler 的作用是将输入的 sku_id_mapping 列组合成一个新的向量列，方便后续的特征处理或模型训练。
          vector:向量     Assembler:汇编程序
     */
//    val assembler=new VectorAssembler()
//      .setInputCols(Array("sku_id_mapping"))
//      .setOutputCol("features")


//    val result01=result.select("user_id_mapping","sku_id_mapping")

    //  对user_id字段也进行索引转换
//    val user_id_indexer=new StringIndexer()
//      .setInputCol("user_id_mapping")
//      .setOutputCol("user_id_indexer")
//      .fit(result01)

    //  对sku_id的数据进行索引转换,即添加索引
//    val indexer=new StringIndexer()
//      .setInputCol("sku_id_mapping")
//      .setOutputCol("sku_id_index")
//      .fit(result01)

//val stringed01=user_id_indexer.transform(result01)
//val stringed=indexer.transform(result)
//
//    // 因为是要转化成一行里面开头是用户id，后面列都是商品id的格式，所以需要将sku_id进行one-hot编码
//    //  one-hot编码需要使用OneHotEncoder,就像映射需要用到StringIndexer
//    //  然后要注意的是必须要先使用fit方法才可以使用后面的transform方法
//    val one_hot_encoder=new OneHotEncoder()
//      .setInputCol("sku_id_index")
//      .setOutputCol("sku_id_onehot")
//      .setDropLast(false)
//      .fit(stringed)
////
//    val one_hot_model=one_hot_encoder.transform(stringed)

//    // 将独热编码后的数据透视 (Pivot)
//    //，实现每个用户对 SKU 的购买情况
//    val pivotData = one_hot_model.groupBy("user_id_mapping")
//      .pivot("sku_id_mapping") // 根据 sku_id_mapping 列创建透视表
//      .agg(first("sku_id_onehot")) // 使用聚合函数
//
//    // 将透视后的数据中的空值填充为 0
//    val matrix = pivotData.na.fill(0.0)
//
//    // 按照用户ID升序排序
//    val sortedMatrix = matrix.orderBy("user_id_mapping")
//
//    // 获取所有 SKU 列名，并选择前5列
//    val skuColumns = matrix.columns.filter(_.startsWith("sku_")).take(5)
//    val selectedColumns = "user_id_mapping" +: skuColumns
//
//    // 使用 select 选择列并展示结果
//    sortedMatrix.select(selectedColumns.map(col): _*).show(5, false)

    //    one_hot_model.select("user_id_indexer","sku_id_onehot")
//      .orderBy("user_id_indexer")
//      .show(5)
//
//    /*
//    (15,[14],[1.0])代表长度为15的向量,也就是总共有15个商品，然后14号位置是非零的,也就是该用户购买了14号商品，
//    所以上面只是一个向量，并没有转换为任务要求的矩阵
//          +---------------+---------------+
//    |user_id_indexer|  sku_id_onehot|
//    +---------------+---------------+
//    |            0.0|(15,[14],[1.0])|
//    |            0.0| (15,[6],[1.0])|
//    |            1.0| (15,[1],[1.0])|
//    |            1.0| (15,[9],[1.0])|
//    |            2.0|(15,[14],[1.0])|
//    +---------------+---------------+
//     */
//
//
//    // 按照 user_id 进行聚合，将 sku_id 作为列名，聚合为 0 或 1 的矩阵
//    val matrix = one_hot_model
//      .groupBy("user_id_indexer") // 按用户进行分组
//      .pivot("sku_id_index") // 将每个 sku_id 的 One-Hot 编码作为列名
//      .agg(max(col("sku_id_onehot"))) // 聚合，生成 0 或 1 的标记
//      .na.fill(0) // 用 0 填充未购买的商品
//
//    // 获取前6列的列名，包括 "user_id_indexer" 和前5个 "sku_id" 列
//    val selectedColumns = "user_id_indexer" +: matrix.columns.slice(1, 6)
//
//    // 使用 select 方法时直接传入 Seq[String] 类型的列名
//    matrix.orderBy("user_id_indexer")
//      .select(selectedColumns.map(col): _*)
//      .show(false)


    //  --------------------------------------------------------------------------------------------------------
    // 对sku_id_mapping进行String类型转换，并添加前缀以避免纯数字列名
//    val transformed = result.select("user_id_mapping", "sku_id_mapping")
//      .withColumn("sku_col", functions.concat(lit("sku_"), col("sku_id_mapping").cast("int")))
//      val number_list=result.select("sku_id_mapping").distinct().map(row => row.get(0).toString.toDouble).collect().toList.sorted
//    println(number_list)
//
//
//    transformed.show
//
//    // 使用pivot进行聚合，标记用户是否购买过每个SKU
//    val matrix = transformed.groupBy("user_id_mapping")
//      .pivot("sku_col")
//      .agg(functions.lit(1.0))
//      .na.fill(0.0) // 未购买的SKU标记为0
//
//
//
//    // 按照用户ID升序排序
//    val sorted_matrix = matrix.orderBy("user_id_mapping")
//
//    // 获取所有SKU列名，并选择前5列（根据实际SKU数量调整）
//    val allColumns = sorted_matrix.columns
//    val selectedColumns = if (allColumns.length > 6) {
//      allColumns.slice(0, 6) // 第一列是user_id_mapping，后5列是SKU
//    } else {
//      allColumns
//    }
//
//    // 使用 select 选择列
//    sorted_matrix.select(selectedColumns.map(col): _*)
//      .show(5, false)
    /*
        +---------------+-----+-----+------+------+------+
    |user_id_mapping|sku_0|sku_1|sku_10|sku_11|sku_12|
    +---------------+-----+-----+------+------+------+
    |0.0            |0.0  |0.0  |0.0   |0.0   |0.0   |
    |1.0            |0.0  |1.0  |0.0   |0.0   |0.0   |
    |2.0            |0.0  |0.0  |0.0   |0.0   |0.0   |
    |3.0            |0.0  |0.0  |0.0   |0.0   |0.0   |
    |4.0            |0.0  |0.0  |0.0   |0.0   |0.0   |
    +---------------+-----+-----+------+------+------+
    only showing top 5 rows
     */

    //  ----------------------------------------------------------------------------------------------------------

    //    正确答案(后面检查发现还是差一点点，因为全是0，没有1)


//    val transformed = result.select("user_id_mapping", "sku_id_mapping")
//      .withColumn("user_id_mapping", col("user_id_mapping").cast("double"))
//      .withColumn("sku_id_mapping", col("sku_id_mapping").cast("int"))
//
//    // 取出商品id，并且排序,方便下面得到题目要求格式的字段list
//    val distinctSkuIds = transformed.select("sku_id_mapping")
//      .distinct()
//      .orderBy("sku_id_mapping") // 按数值顺序排序
//
//    //  getAs[Int] 是一个方法，用于获取指定列名（这里是 "sku_id_mapping"）的值，并将其转换为指定的类型（这里是 Int）。
//    //  然后collect()返回是一个row对象
//    val skuIdList = distinctSkuIds.collect().map(row => "sku_" + row.getAs[Int]("sku_id_mapping")).toList
//
//    transformed.withColumn("sku_id_mapping", col("sku_id_mapping").cast("double"))
//
//    // groupBy("user_id_mapping"):根据字段分组，同一个用户id的数据会全部聚合在一起
//    //  .pivot("sku_id_mapping", skuIdList): 将 sku_id_mapping 列的值转为列名，并根据 skuIdList 指定的 SKU ID 列表创建新的列。
//    //  通过pivot函数，会把sku_id展开为多个列名，每个列名代表一个sku
//    //  .agg(functions.lit(1.0)): 对每个新的列应用聚合函数，这里使用 lit(1.0)，意味着每次购买都会标记为 1.0。
//    //  .na.fill(0.0): 用于填充缺失值，将未购买的 SKU 标记为 0.0。
//    val matrix = transformed.groupBy("user_id_mapping")
//      .pivot("sku_id_mapping", skuIdList) // 按顺序列出sku列
//      .agg(functions.max(when(col("sku_id_mapping").isNotNull, 1.0).otherwise(0.0)))  // 根据是否购买填充 1 或 0
//      .na.fill(0.0) // 未购买的SKU标记为0
//
//    // 按照用户ID升序排序
//    val sorted_matrix = matrix.orderBy("user_id_mapping")
//
//    // 获取所有SKU列名，并选择前5列
//    val selectedColumns = "user_id_mapping" :: skuIdList.take(10) // 选择前5个SKU列
//
//
//    println("---------------第一行前5列结果展示为---------------")
//    // 使用 select 选择列
//    sorted_matrix.select(selectedColumns.map(col): _*)
//      .show(5000, false)

    /*

         +---------------+-----+-----+-----+-----+-----+
    |user_id_mapping|sku_0|sku_1|sku_2|sku_3|sku_4|
    +---------------+-----+-----+-----+-----+-----+
    |0              |0.0  |0.0  |0.0  |0.0  |0.0  |
    |1              |0.0  |0.0  |0.0  |0.0  |0.0  |
    |2              |0.0  |0.0  |0.0  |0.0  |0.0  |
    |3              |0.0  |0.0  |0.0  |0.0  |0.0  |
    |4              |0.0  |0.0  |0.0  |0.0  |0.0  |
    +---------------+-----+-----+-----+-----+-----+
     */




    //  -------------------------------------------------------------------------------------------------------
    //  按照user_id进行聚合
//    val matrix=one_hot_model
//      .groupBy("user_id_mapping")
//      .pivot("sku_id_mapping")
//      .agg(functions.sum("sku_id_onehot"))
//      .na.fill(0)
//      .orderBy("user_id_mapping")
//
//    //  展示矩阵第一行前五列数据
//    matrix.select("user_id_mapping",matrix.columns.slice(2,7): _*).show()
//
//
//
//
//    // 创建Pipeline来执行上述操作
//    val pipeline=new Pipeline()
//      .setStages(Array(indexer,one_hot_encoder))
//
//    //  选择进行训练的对象
//    val model=pipeline.fit(result01)
//    //  开始训练
//    val one_hot_result=model.transform(result01)
//
//    one_hot_result.show
//
//    // 定义一个 UDF 来将 SparseVector 转换为 DenseVector
//    val toDense = udf((v: SparseVector) => Vectors.dense(v.toArray))
//
//    // 将 sku_id_onehot 列中的 SparseVector 转换为 DenseVector
//    val transformedDf = one_hot_result.withColumn("sku_id_onehot_dense", toDense(col("sku_id_onehot")))
//
//    //  获取sku的数量
//    val sku_count=transformedDf.select("sku_id_onehot").distinct.count()
//
//    //  创建用户-商品矩
//    val aggExprs = (0 until sku_count.toInt).map(i =>
//      max(when(col("sku_id_onehot.values").getItem(i) === 1.0, 1).otherwise(0)).alias(s"sku_$i")
//    )
//
//    val purchaseDf =one_hot_result.groupBy("user_id").agg(aggExprs.head, aggExprs.tail: _*)

    //  -------------------------------------------------------------------------------------------

    // 使用OneHotEncoder进行one-hot编码
    val oneHotEncoder = new OneHotEncoder()
      .setInputCol("sku_id_mapping")
      .setOutputCol("sku_id_onehot")
      .setDropLast(false) // 保留最后一个值
      .fit(result)

    val oneHotResult = oneHotEncoder.transform(result)

    // 使用pivot对sku_id进行透视操作，将其转为列
    val skuList = oneHotResult.select("sku_id_mapping").distinct().collect().map(row => row.getDouble(0).toInt)
    val skuColumns = skuList.map(skuId => "sku_" + skuId)

    val pivotData = oneHotResult.groupBy("user_id_mapping")
      .pivot("sku_id_mapping", skuColumns)
      .agg(functions.count(col("sku_id_mapping"))) // 使用max聚合，保留one-hot值
      .na.fill(0.0) // 填充空值为0

    // 显示结果，展示user_id以及前5个sku列
    val selectedColumns = "user_id_mapping" +: skuColumns.take(5)

    pivotData.select(selectedColumns.map(col): _*)
      .orderBy("user_id_mapping")
      .show(5, false)


    spark.close()

  }

}
